package com.shujia.dws

import com.shujia.utils.SparkTool
import org.apache.spark.sql._
import org.apache.spark.storage.StorageLevel

object SpacetimeAdjoint extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    //1、读取位置融合表
    val mergeLocation: DataFrame = spark
      .table("dwi.dwi_res_regn_mergelocation_msk_d")
      .where($"day_id" === day_id)


    //2、使用jdb读取mysql中的确诊列表和密接列表

    //确诊列表
    val confirmed: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable", "crm.confirmed")
      .option("user", "root")
      .option("password", "123456")
      .load()

    /**
      * 1、获取确诊人员的行动轨迹
      */

    val confirmedLocation: DataFrame = mergeLocation
      .join(confirmed.hint("broadcast"), "mdn")
      .select($"mdn" as "confirmed_mdn", $"start_date" as "confirmed_start_date", $"longi" as "confirmed_longi", $"lati" as "confirmed_lati", $"county_id")

    /**
      * 时空伴随着的判断条件
      * 1、时间差一小时
      * 2、距离差500米
      *
      */


    //按照区县关联
    val filterDF: DataFrame = mergeLocation
      .join(confirmedLocation.hint("broadcast"), "county_id")
      //排除确诊人员
      .where($"confirmed_mdn" =!= $"mdn")
      //1、计算用户和确诊人员数据相差的时间
      .withColumn("diff", abs(unix_timestamp($"confirmed_start_date", "yyyyMMddHHmmss") - unix_timestamp($"start_date", "yyyyMMddHHmmss")))
      //过滤数据
      .where($"diff" < 3600)
      //计算距离
      .withColumn("length", calculateLength($"confirmed_longi", $"confirmed_lati", $"longi", $"lati"))
      //过滤数据
      .where($"length" < 500)
      .select($"mdn"
        , $"start_date"
        , $"end_date"
        , $"county_id"
        , $"longi"
        , $"lati"
        , $"bsid"
        , $"grid_id")


    filterDF.persist(StorageLevel.MEMORY_AND_DISK_SER)

    //保存密切接触者规则
    filterDF
      .write
      .format("csv")
      .mode(SaveMode.Overwrite)
      .option("sep", "\t")
      .save(s"/daas/motl/dws/spacetime_adjoint/day_id=$day_id")


    //取出密切接触者手机号
    filterDF
      .select("mdn")
      .distinct()
      .write
      .format("csv")
      .mode(SaveMode.Overwrite)
      .option("sep", "\t")
      .save(s"/daas/motl/dws/spacetime_adjoint_mdns/day_id=$day_id")


  }
}
